package demo.spark.core

import java.util.concurrent.atomic.LongAccumulator

import demo.spark.model.UserInfo
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable
import scala.reflect.ClassTag

/**
 * 对于Spark而言，Spark提供一套适用于分布式，基于流式处理与批处理并兼容性能的运算框架，
 * Spark-Core模块定义了Spark核心概念，例如SparkConf, SparkContext, RDD，算子以及action，
 * RDD: 弹性分部署数据集 (resilient distributed dataset) 是针对分布式大数据分片存储的的一种可行的 "数据投影" 方案。
 * 算子: 在大数据计算或流式计算中，可能发起一次数据计算包含多个map过程，每个map过程实际就是一段描述数据在当前阶段convert的逻辑
 *      以及数据流向下一阶段的表现形式，因此 算子 只是在任务定义阶段的逻辑描述，真正的执行是在触发到任务action时进行阶段性逻辑运算！
 * action: 运算行为，一次有效的运算必须1个action，可以包含0个或多个算子，action是真正触发 算子所定义逻辑运算 的点火锁，
 *      在整个算子逻辑执行的过程中，实际上是 由Spark构建的 "数据结构化投影RDD" 在不同运算单元中的转换！
 *
 * NOTICE: RDD不是具体的数据，Spark不会将每次运算所使用的数据加载到内存！而是分布式文件系统中数据的结构化投影，
 *         用于配合算子描述数据的转换逻辑！action是在所有算子逻辑执行结束后的结果集归并，该结果集可被Spark收集！
 *
 *
 * <!-- 以下摘自Spark官网:
 *  https://spark.apache.org/docs/latest/rdd-programming-guide.html#resilient-distributed-datasets-rdds
 * -->
 *
 * RDD 操作:
 * RDD 支持两种类型的操作：transforms，它从现有的数据集创建一个新的数据集，以及actions，
 * 它在对数据集运行计算后返回一个值给驱动程序。例如，map是一个转换，它通过一个函数传递每
 * 个数据集元素并返回一个表示结果的新 RDD。另一方面，reduce是使用某个函数聚合RDD的所有元
 * 素并将最终结果返回给驱动程序的操作（尽管也有一个并行reduceByKey返回分布式数据集）。
 *
 * Spark 中的所有转换都是惰性的，因为它们不会立即计算结果。相反，他们只记住应用于某些基本数据集（例如文件）的转换。
 * 仅当操作需要将结果返回到驱动程序时才计算转换。这种设计使 Spark 能够更高效地运行。例如，我们可以意识到通过创建的
 * 数据集map将在 a 中使用，reduce并且仅将 的结果返回reduce给驱动程序，而不是更大的映射数据集。
 *
 * 默认情况下，每次在其上运行操​​作时，每个转换后的 RDD 可能会重新计算。但是，你也可以坚持使用内存中的RDD persist
 * （或cache）方法，在这种情况下，星火将保持元件周围的群集上更快访问你查询它的下一次。还支持在磁盘上持久化 RDD，或跨多个节点复制。
 */
object ContextDemo {

  def main(args: Array[String]): Unit = {
    //创建基于local环境的spark上下文配置.
    var conf: SparkConf = new SparkConf()
      .setAppName("ContextDemo")
      .setMaster("local");

    //创建spark上下文.
    var sparkContext: SparkContext = new SparkContext(conf);

    /*
    //spark上下文并行化一个数组 Array(1,2,3,4,5) => resilient distributed dataset (RDD) 弹性分布式数据集投影.
    val sum:Int = sparkContext.parallelize(Array(1,2,3,4,5))
      //对内存中的RDD执行求和的action.
      .reduce((left: Int, right: Int) => left + right);
    println(sum);
    */



    /*
    //spark 允许输入源为本地文件系统目录 / 分布式的文件系统，例如 hdfs://icasue111:9090/dir/text.txt, 并投影成RDD/内存模型进行运算.
    sparkContext.textFile("/Users/icasue/Desktop/lbs-server-plugins/icasue-plugins-demos/icasue-demo-spark/local_text/user.txt")
      .map(line => line.split(","))
      .map(lineItems => {
        val id = lineItems(0).toInt;
        val name = lineItems(1).trim;
        val age = lineItems(2).toInt;
        val hobby:Array[String] = lineItems(3).split("-");
        val week:Map[String,String] = lineItems(4)
            .split("-")
            .map(kv => (kv.split(":")(0), kv.split(":")(1)))
            .toMap[String,String];
        new UserInfo(id,name,age,hobby,week);
      })
      //cluster模式下spark的计算是分片聚合的，简单调用foreach无法在运算节点的驱动上打印相关日志，需要使用 rdd.collect().foreach(println())
      //rdd.collect() 本质上是将集群中不通运算节点当前阶段的rdd收集到驱动程序，后调用foreach，可能存在内存问题，因此最合理的方式是
      // rdd.collect().take(topNum).foreach(println()) 保证收集节点的内存开销不会太大.
      .collect()
      .take(100)
      .foreach(item => println(item.format()));
     */




    /*
    val str_un_know: String = "UN_KNOW";
    var ary:Array[(String,Seq[Int])] =  Array(
      ("qiaohang",Array(1,2,3)),
      ("qiaohang",Array(1,2,3)),
      ("zhouzhou",Array(4,5,6)),
      ("qiyan",Array(7,8,9)),
      ("",Array(1,3,5)),
      ("SUM_ZERO",Array(0)),
      ("SUM_NULL",null:Array[Int])
    );
    val groupSumAggregate: mutable.HashMap[String,Int] = sparkContext.makeRDD(ary)
      .aggregate[mutable.HashMap[String, Int]] (new mutable.HashMap[String,Int]{{ put(str_un_know,0) }}) ((result, item) => {
        if (item._1.equals("")) {
          if (item._2 == null || item._2.size == 0) None else
            result.put(str_un_know, result.getOrElse(str_un_know, 0) + item._2.reduce((left, right) => left + right));
        } else {
          if (item._2 == null || item._2.size == 0) result.put(item._1, result.getOrElse(item._1, 0)) else
            result.put(item._1, result.getOrElse(item._1, 0) + item._2.reduce((left, right) => left + right));
        }
        result;
      }, (defaultForResult, partitionN) => {
        // 用于合并不同分区中的处理结果，local环境下只有一个分区！
        partitionN.foreach(entry => defaultForResult.+(entry))
        partitionN
      });

    // Map(qiyan -> 24, zhouzhou -> 15, qiaohang -> 12, SUM_NULL -> 0, UN_KNOW -> 9, SUM_ZERO -> 0)
    println(groupSumAggregate)
    */




    /*
    class SortModel(implicit asc:Boolean = true) extends scala.math.Ordering[(String,Int,Int)]{
      override def compare(x: (String, Int, Int), y: (String, Int, Int)): Int = {
        var compare = 0;
        if((compare = x._1.compareTo(y._1)) != 0)
          None;
        else if ((compare = x._2.compareTo(x._2)) != 0)
          None;
        else
          compare = x._3.compareTo(x._3)
        if(asc) compare * 1 else -1 * compare;
      }
    }

    sparkContext.makeRDD(ary)
      .sortBy[(String,Int,Int)](item => (item._1,item._2,item._3))(new SortModel, ClassTag.apply(Class[(String,Int,Int)]))
      .foreach(item => println(item))

    sparkContext.makeRDD(ary)
      .saveAsTextFile("/Users/icasue/Desktop/lbs-server-plugins/icasue-plugins-demos/icasue-demo-spark/local_text/save")
    */


    /*
    val value:Broadcast[List[Int]] = sparkContext.broadcast(List(1, 2, 3));

    var ary:Array[(String,List[Int])] = Array(
      ("aQiao",List(1,2)),
      ("bQiao",List(2,3)),
      ("bQiao",List(2,3)),
      ("bZhou",List(3,4)),
      ("cYan",List(5,6))
    );
    ary
      .map(item => (item._1, List() ::: item._2 ::: value.value))
      .map(item => (item._1, List() ::: item._2 ::: value.value))
      //释放执行程序上的广播资源, 下次执行程序使用时将重新广播!
      .map(item => {value.unpersist(); item})
      .map(item => (item._1, List() ::: item._2 ::: value.value))
      //销毁执行程序上的广播资源, 广播变量将无法使用！map操作异常 (但是这里存在问题是因为广播变量被map阶段销毁多次！)
      .map(item => {value.destroy(); item})
      //.map(item => (item._1, List() ::: item._2 ::: value.value))
      .foreach(item => println(item))
     */


    /*
    val rdd = sparkContext.parallelize(Array(("casue",1), ("casue",1), ("zhouzhou",3)), 1)
    val rdd2 = sparkContext.parallelize(Array(("casue","eat"), ("zhouzhou","sleep")), 1)
    //join操作按Tuple2的第一维分组，结果为: Tuple2[String,(Int,String)]
    val value = rdd.distinct().join(rdd2.distinct());
    value.foreach(item => println(item))
     */

  }

}
